package com.kun.lagou.seckillservice.consumer;

import com.kun.lagou.order.api.dto.OrderInfo;
import com.kun.lagou.seckillservice.dao.SeckillOrderDao;
import com.kun.lagou.seckillservice.model.SeckillOrder;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.spring.annotation.RocketMQTransactionListener;
import org.apache.rocketmq.spring.core.RocketMQLocalTransactionListener;
import org.apache.rocketmq.spring.core.RocketMQLocalTransactionState;
import org.springframework.messaging.Message;
import org.springframework.transaction.support.TransactionTemplate;

/**
 *
 * @author Administrator
 */
@Slf4j
@RocketMQTransactionListener
@RequiredArgsConstructor
public class CreateOrderRocketMQTransactionListener implements RocketMQLocalTransactionListener {
    private final TransactionTemplate transactionTemplate;
    private final SeckillOrderDao seckillOrderDao;

    //1. 发送半消息

    /**
     * 用来执行本地事务
     * @param msg
     * @param arg
     * @return
     */
    @Override
    public RocketMQLocalTransactionState executeLocalTransaction(Message msg, Object arg) {
        //2. 执行本地事务
        CreateSeckillOrderConsumer.CreateSeckillOrderMessage message = (CreateSeckillOrderConsumer.CreateSeckillOrderMessage) arg;
        var seckillOrder = transactionTemplate.execute(s -> {
            if(seckillOrderDao.findByUserIdAndActivityId(message.userId, message.acitvityId).isEmpty()) {
                //也可以用乐观锁
                return seckillOrderDao.save(SeckillOrder.builder()
                        .userId(message.userId)
                        .activityId(message.acitvityId)
                        .build());
            }
            return null;
        });
        return seckillOrder == null ? RocketMQLocalTransactionState.ROLLBACK : RocketMQLocalTransactionState.COMMIT;
        //3. 提交半消息
        //异常情况：提交失败，这时候消息还是处于prepare状态，mq帮我们回调下面的代码checkLocalTransaction
    }

    /**
     * 本地事务的检查接口,即MQServer没有收到二次确认消息时调用的方法
     * 去检查本地事务到底有没有成功
     * @param msg
     * @return
     */
    @Override
    public RocketMQLocalTransactionState checkLocalTransaction(Message msg) {
        OrderInfo orderInfo = (OrderInfo) msg.getPayload();
        var userId = orderInfo.getUserId();
        var activityId = (Long)orderInfo.getParams().get("activityId");
        return seckillOrderDao.findByUserIdAndActivityId(userId, activityId).isPresent() ?
                RocketMQLocalTransactionState.COMMIT : RocketMQLocalTransactionState.ROLLBACK;
    }
}
